package com.atguigu.gmall.apps

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

import com.alibaba.fastjson.JSON
import com.atguigu.gmall.Beans.OrderInfo
import com.atguigu.gmall.utils.Constant.KafkaConstant
import com.atguigu.gmall.utils.MykafkaUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}


/**
  *
  * 处理日销售额GMV
  *
  * 1、通过sparkstreaming从kafka拉取数据
  * 2、将获取的json数据转化为GMV对象
  * 2、在phoenix中创建表，
  * 3、将数据存储到hbase中
  *
  *
  */
object GMVApp extends baseApp {

  override val AppName: String = "GMV"
  //5s拉取一次
  override val time: Int = 5


  def main(args: Array[String]): Unit = {

    runApp({

      //1，通过sparkstreaming消费kafka数据

      val ds: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(KafkaConstant.KAFKA_TOPIC_ORDER_INFO,streamingContext)

      //2、将数据转化为orderinfo对象
      val ds1 = ds.map(record=>{

        val orderinfo: OrderInfo = JSON.parseObject(record.value(),classOf[OrderInfo])


        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formatter2 = DateTimeFormatter.ofPattern("yyyy-MM-dd")

        //将string的日期格式转化为LocalDateTime类型
        val time: LocalDateTime = LocalDateTime.parse(orderinfo.create_time,formatter)

        //获取创建订单的日期
        orderinfo.create_date = time.format(formatter2)

        //获取创建订单的时间
        orderinfo.create_hour = LocalDateTime.parse(orderinfo.create_time,formatter).getHour.toString

        //对手机号前三后四脱敏
        orderinfo.consignee_tel = orderinfo.consignee_tel.replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2");
        //将转化的对象返回
        orderinfo


      })

      //将数据通过phoenix存储在HBase中
      WriteToHbase(ds1)

    })



  }


  //将数据写入到HBase中
  def WriteToHbase(ds: DStream[OrderInfo]): Unit = {

    //使用了隐式转换，可以直接使用rdd调用phoenix的方法
    import org.apache.phoenix.spark._

    ds.foreachRDD(rdd=>{

      //将数据存储到HBase中
      rdd.saveToPhoenix("GMALL2021_ORDER_INFO",
        Seq("ID","PROVINCE_ID", "CONSIGNEE", "ORDER_COMMENT", "CONSIGNEE_TEL",
            "ORDER_STATUS", "PAYMENT_WAY", "USER_ID","IMG_URL", "TOTAL_AMOUNT",
            "EXPIRE_TIME", "DELIVERY_ADDRESS", "CREATE_TIME","OPERATE_TIME",
            "TRACKING_NO","PARENT_ORDER_ID","OUT_TRADE_NO", "TRADE_BODY",
            "CREATE_DATE", "CREATE_HOUR"),
        HBaseConfiguration.create(),
        Some("hadoop102,hadoop103,hadoop104:2181")
      )
    })

  }

}
